package com.ndhutu.ee.util.ali.rocketmq;

import com.aliyun.openservices.ons.api.*;

import java.util.Properties;

/**
 * @Author Shane
 * @Date 2019/9/19 23:39
 */
public class RocketMQUtil {
    /**
     *  初始化
     * @Author Shane
     * @Date 2019/9/19 23:40
     * @param 
     * @return 
     */
    public static Properties initProperties(RocketMQConfig config){
        Properties properties = new Properties();

        properties.put(PropertyKeyConst.GROUP_ID, config.getGroupId());
        properties.put(PropertyKeyConst.AccessKey, config.getAccessKeyId());
        properties.put(PropertyKeyConst.SecretKey, config.getAccessKeySecret());
        properties.put(PropertyKeyConst.NAMESRV_ADDR, config.getNamesrvAddr());

        return properties;
    }

    /**
     * 发送消息
     * @Author Shane
     * @Date 2019/9/2011:42
     * @param tag   消息标签
     * @param msg   消息体
     * @return
     **/
    public static void sendMessage(RocketMQConfig config, String tag, String msg) {
        Producer producer = ONSFactory.createProducer(initProperties(config));
        // 在发送消息前，必须调用 start 方法来启动 Producer，只需调用一次即可
        producer.start();

        Message message = new Message(config.getTopic(), tag, msg.getBytes());

        // 设置代表消息的业务关键属性，请尽可能全局唯一，以方便您在无法正常收到消息情况下，可通过消息队列 MQ 控制台查询消息并补发
        // 注意：不设置也不会影响消息正常收发
//        message.setKey("ORDERID_100");

        SendResult result = producer.send(message);

        // 在应用退出前，可以销毁 Producer 对象
        // 注意：如果不销毁也没有问题
//        producer.shutdown();
    }

    /**
     * 消费消息
     * @Author Shane
     * @Date 2019/9/2011:42
     * @param
     * @return
     **/
    public static void subMessage(RocketMQConfig config){
        Consumer consumer = ONSFactory.createConsumer(initProperties(config));

        consumer.subscribe(config.getTopic(), "*", new MessageListener() {
            public Action consume(Message message, ConsumeContext consumeContext) {

                return Action.CommitMessage;
            }
        });

        consumer.start();
    }

}
